/**
 * Copyright(c) 2013 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-7-29 下午1:14 Created by dwzhang
 */
package com.jade.framework.base.comm;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.logging.Log;

/**
 * 抽象的通信客户端。
 *
 * @author dwzhang
 * @version 1.0 14-7-29 下午1:14
 */
public abstract class AbstractClient<Channel>
        implements CommClient<Channel>, Runnable
{
    protected Protocol protocol;
    protected ScheduledExecutorService executorService;
    protected ProtocolTransport transport;
    protected ProtocolSession session;
    protected Log log;

    public void setProtocol(Protocol protocol)
    {
        this.protocol = protocol;
    }

    public Protocol getProtocol()
    {
        return protocol;
    }

    @PostConstruct
    public void init()
    {
        log = protocol.getLog();
        ProtocolConfiguration configuration = protocol.getConfiguration();
        if (configuration.isAutoConnect() && configuration.isLongTerm()) {
            //长连接协议的情况，自动建立连接。
            executorService = Executors.newSingleThreadScheduledExecutor();
            executorService.scheduleAtFixedRate(this, 5000, 10000, TimeUnit.MILLISECONDS);
        }
    }

    @PreDestroy
    public void destroy()
    {
        close();
        if (executorService != null) {
            executorService.shutdownNow();
            executorService = null;
        }
    }

    public void run()
    {
        ProtocolConfiguration configuration = protocol.getConfiguration();
        try {
            if (!isAlive()) {
                createConnection();
            }
        }
        catch (Throwable t) {
            log.error("Can't create connection to " + configuration.getHost() + ":" + configuration.getPort(), t);
        }
    }

    protected synchronized void createConnection()
            throws CommException
    {
        if (isAlive()) {
            return;
        }
        doCreateConnection();
    }

    protected abstract void doCreateConnection()
            throws CommException;


    public synchronized boolean isAlive()
    {
        return transport != null && transport.isAlive();
    }

    public void send(Object message)
            throws CommException
    {
        createConnection();
        transport.send(message);
    }

    public ResponseMessage sendRequest(RequestMessage request)
            throws CommException, InterruptedException
    {
        createConnection();
        return transport.sendAndGetResponse(request, 30000);
    }

    public ResponseMessage sendRequest(RequestMessage request, int timeout)
            throws CommException, InterruptedException
    {
        createConnection();
        return transport.sendAndGetResponse(request, timeout);
    }

    public void close()
    {
        if (transport != null) {
            if (transport.isAlive()) {
                transport.close();
            }
            transport = null;
            session = null;
        }
    }

    public void addSession(ProtocolSession session)
    {
        this.session = session;
    }

    public ProtocolSession getSession(Channel channel)
    {
        return session;
    }
}
